package com.zxinlee.m2;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

/**
 * 顺序消息
 * <br>
 * Rocketmq 顺序消息的基本原理：
 * <p>
 * 同一组有序的消息序列，会被发送到同一个队列，按照 FIFO 的方式进行处理
 * 一个队列只允许一个消费者线程接收消息，这样就保证消息按顺序被接收
 * <p>
 * 下面以订单为例：
 * <br>
 * 一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中。
 * 消费时，从同一个队列接收同一个订单的消息。
 */
public class Producer {

    //订单消息
    // 以下消息, 相同id的消息按顺序发送到同一个队列,
    // 消费时也从同一个队列按顺序消费
    static String[] msgs = {
            "15103111039,创建",
            "15103111065,创建",
            "15103111039,付款",
            "15103117235,创建",
            "15103111065,付款",
            "15103117235,付款",
            "15103111065,完成",
            "15103111039,推送",
            "15103117235,完成",
            "15103111039,完成"
    };

    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        //创建生产者
        DefaultMQProducer producer02 = new DefaultMQProducer("producer02");

        //设置 name server
        producer02.setNamesrvAddr("192.168.64.141:9876");

        //启动
        producer02.start();

        //发送消息
        for (String msg : msgs) {
            // msg -- "15103111039,创建"  --> 从消息中取出订单编号
            Long orderId = Long.valueOf(msg.split(",")[0]);
            //封装消息
            Message message = new Message("Topic02", msg.getBytes());

            //队列选择器
            MessageQueueSelector messageQueueSelector = new MessageQueueSelector() {
                /**
                 *
                 * @param mqs 服务器端的队列列表
                 * @param msg 正要发送的消息
                 * @param arg 选择依据
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Long orderId = (Long) arg;
                    int index = (int) (orderId % mqs.size());
                    return mqs.get(index);
                }
            };
            //producer02.send(message,队列选择器,选择依据);
            SendResult result = producer02.send(message, messageQueueSelector, orderId);

            System.out.println(orderId + "-->" + result.getMessageQueue().getQueueId());
        }

    }
}
